use byteorder::LittleEndian;
use flate2::read::ZlibDecoder;
use log::{debug, error};
use serde_json::Value;
use std::collections::HashMap;
use std::io::{Read, Write};
use std::net::SocketAddr;
use std::sync::MutexGuard;
use std::thread;
use std::time::{Duration, Instant};
use tokio::io::AsyncReadExt;
use tokio::io::AsyncWriteExt;
use tokio::io::{self};
use tokio::net::{TcpSocket, TcpStream};
use tokio::sync::Mutex; // Assuming we're using serde_json for JSON handling

use byteorder::ByteOrder;
use std::convert::TryFrom;

use crate::helper;
const CONNECT_TIMEOUT: u64 = 5000; // 5 seconds
const RECV_HEADER_LEN: usize = 0x10;
const DEFAULT_HEARTBEAT_INTERVAL: f64 = 10.0;
// Define constants similar to the Python code
const RSP_HEADER_LEN: usize = 0x10; // Adjust based on the actual header length

/// Define a struct that represents the response header
#[derive(Debug)]
struct ResponseHeader {
    version: u32,
    msg_type: u16,
    reserved: u16,
    zip_size: i32,
    unzip_size: i32,
}
// Define custom errors
#[derive(Debug)]
pub enum ConnectionError {
    Timeout,
    Io(std::io::Error),
}

impl From<std::io::Error> for ConnectionError {
    fn from(err: std::io::Error) -> Self {
        ConnectionError::Io(err)
    }
}

// Define retry strategy
#[derive(Debug, Clone)]
struct RetryStrategy {
    pub intervals: Vec<f64>, // Intervals in seconds
}

impl RetryStrategy {
    fn new(intervals: Vec<f64>) -> Self {
        RetryStrategy { intervals }
    }

    fn next_interval(&mut self) -> Option<f64> {
        self.intervals.pop() // Get the next interval and remove it from the list
    }
}

// Define the TrafficStatSocket struct
struct TrafficStatSocket {
    stream: TcpStream,
    send_pkg_num: u64,
    recv_pkg_num: u64,
    send_pkg_bytes: u64,
    recv_pkg_bytes: u64,
    first_pkg_send_time: Option<Instant>,
    last_api_send_bytes: u64,
    last_api_recv_bytes: u64,
}

impl TrafficStatSocket {
    pub fn new(stream: TcpStream) -> Self {
        TrafficStatSocket {
            stream,
            send_pkg_num: 0,
            recv_pkg_num: 0,
            send_pkg_bytes: 0,
            recv_pkg_bytes: 0,
            first_pkg_send_time: None,
            last_api_send_bytes: 0,
            last_api_recv_bytes: 0,
        }
    }

    // Add methods for sending and receiving data, updating stats, etc.
}

// Define the BaseSocketClient struct
pub struct DefaultRetryStrategy;

impl DefaultRetryStrategy {
    pub fn gen(&self) -> impl Iterator<Item = f64> {
        // 默认重试4次 ... 时间间隔如下
        std::iter::once(0.1)
            .chain(std::iter::once(0.5))
            .chain(std::iter::once(1.0))
            .chain(std::iter::once(2.0))
    }
}

pub struct BaseSocketClient {
    pub need_setup: bool,
    pub lock: Option<Mutex<()>>,
    pub client: Option<TcpStream>, // Assuming TcpStream is defined elsewhere; you might need to replace it with your actual type
    pub heartbeat: bool,
    pub heartbeat_thread: Option<thread::JoinHandle<()>>,
    pub heartbeat_interval: f64, // in seconds
    pub last_ack_time: Instant,
    pub last_transaction_failed: bool,
    pub ip: Option<String>,
    pub port: Option<u16>,

    pub auto_retry: bool,
    pub retry_strategy: DefaultRetryStrategy,
    pub raise_exception: bool,
}

impl BaseSocketClient {
    pub fn new(
        multithread: bool,
        heartbeat: bool,
        auto_retry: bool,
        raise_exception: bool,
    ) -> Self {
        BaseSocketClient {
            need_setup: true,
            lock: if multithread || heartbeat {
                Some(Mutex::new(()))
            } else {
                None
            },
            client: None,
            heartbeat,
            heartbeat_thread: None,
            heartbeat_interval: 10.0, // Default 10 seconds for a heartbeat
            last_ack_time: Instant::now(),
            last_transaction_failed: false,
            ip: None,
            port: None,
            auto_retry,
            retry_strategy: DefaultRetryStrategy,
            raise_exception,
        }
    }

    pub async fn connect(
        &mut self,
        ip: Option<&str>,
        port: Option<u16>,
        // timeout: u64,
    ) -> Result<(), ConnectionError> {
        let ip = ip.unwrap_or("101.227.73.20");
        let port = port.unwrap_or(7709);

        // let timeout_duration = Duration::from_millis(timeout);

        println!("connecting to server: {}:{}", ip, port);

        let socket: TcpStream = TcpStream::connect(format!("{ip}:{port}")).await?;

        println!("Connected to server: {}", socket.peer_addr().unwrap());

        self.client = Some(socket);

        Ok(())
    }

    pub async fn send(&mut self, pkg: &Vec<u8>) -> Result<(), io::Error> {
        // Ensure a client is connected
        if self.client.is_none() {
            return Err(io::Error::new(
                io::ErrorKind::NotConnected,
                "No connection established",
            ));
        }

        // Get a mutable reference to the TcpStream
        if let Some(client) = self.client.as_mut() {
            // Attempt to send the data
            client.write_all(&pkg).await?;

            // Flush the stream to ensure all data is sent
            client.flush().await?;

            Ok(())
        } else {
            Err(io::Error::new(io::ErrorKind::Other, "Client is not set"))
        }
    }

    pub async fn recv(&mut self) -> Result<Vec<u8>, io::Error> {
        // Ensure a client is connected
        if self.client.is_none() {
            return Err(io::Error::new(
                io::ErrorKind::NotConnected,
                "No connection established",
            ));
        }

        // Get a mutable reference to the TcpStream
        let client = self.client.as_mut().unwrap();

        // Read the response header
        let mut head_buf = [0; RSP_HEADER_LEN];
        let bytes_read = client.read(&mut head_buf).await?;

        if bytes_read != RSP_HEADER_LEN {
            return Err(io::Error::new(
                io::ErrorKind::UnexpectedEof,
                "Header length does not match expected size",
            ));
        }

        //                 _, _, _, zipsize, unzipsize = struct.unpack("<IIIHH", head_buf)
        let (_pkg_type, _unknown_1, _unknown_2, zipsize, unzipsize) = (
            LittleEndian::read_u32(&head_buf[0..4]),
            LittleEndian::read_u32(&head_buf[4..8]),
            LittleEndian::read_u32(&head_buf[8..12]),
            LittleEndian::read_u16(&head_buf[12..14]), // Adjusted for `H`
            LittleEndian::read_u16(&head_buf[14..16]), // Adjusted for `H`
        );

        println!(
            "\nzipsize={} unzipsize={} head_buf={}",
            zipsize,
            unzipsize,
            hex::encode(&head_buf)
        );

        // Read the response body
        let mut body_buf = vec![0; 0];
        let mut total_read = 0;
        while total_read < zipsize as usize {
            let mut buf = vec![0; zipsize as usize - total_read];
            let len_buf = client.read(&mut buf).await?;
            total_read += len_buf;
            body_buf.extend_from_slice(&buf[..len_buf]);

            if len_buf == 0 {
                return Err(io::Error::new(
                    io::ErrorKind::UnexpectedEof,
                    "接收数据体失败服务器断开连接".to_string(),
                ));
            }
        }

        if total_read != zipsize as usize {
            return Err(io::Error::new(
                io::ErrorKind::UnexpectedEof,
                "Body length does not match expected size",
            ));
        }
        if zipsize == unzipsize {
            println!("不需要解压");
        } else {
            println!("需要解压");
        }
        // Convert each byte in `body_buf` to a two-digit hex string and collect them into a Vec<String>
        // let hex_values: Vec<String> = body_buf.iter().map(|&b| format!("{:02x}", b)).collect();
        // println!("body_buf={}", hex_values.join(" "));

        // Decompress the body if necessary
        let decompressed_body = if zipsize == unzipsize {
            body_buf // No decompression needed
        } else {
            let mut decoder = ZlibDecoder::new(body_buf.as_slice());
            let mut decompressed = Vec::new();
            decoder.read_to_end(&mut decompressed)?;
            decompressed
        };

        let hex_values: Vec<String> = decompressed_body
            .iter()
            .map(|&b| format!("{:02x}", b))
            .collect();

        // println!("\ndecompressed_body= {:?}", hex_values.join(" "));

        Ok(decompressed_body)
    }

    pub async fn disconnect(&mut self) -> io::Result<()> {
        let client = self
            .client
            .as_mut()
            .ok_or_else(|| io::Error::new(io::ErrorKind::Other, "Client not connected"))?;

        // Acquire the lock if it exists synchronously
        if let Some(lock) = &self.lock {
            let _lock = lock.lock();
        }

        // Gracefully shut down the write side of the connection
        client.shutdown().await?;

        // Optionally, you might want to wait for the read side to complete or close it as well
        // client.shutdown(std::net::Shutdown::Both).await?;

        // Clear the client to indicate it's disconnected
        self.client.take();

        Ok(())
    }

    // Add other methods for heartbeat, retries, and traffic stats
}

#[cfg(test)]
mod test {
    use std::{
        thread::{self, Thread},
        time::Duration,
    };

    use crate::{
        base_socket_client::BaseSocket
        get_history_transaction_data::GetHistoryTransactionData,
        setup_cmd1::{self, SetupCmd1},
        setup_cmd2::SetupCmd2,
        setup_cmd3::SetupCmd3,
    };
    use byteorder::{ByteOrder, LittleEndian};
    use log::{debug, error};

    #[tokio::test]
    async fn test_connect() {
        // Example usage of BaseSocketClient
        let mut client = BaseSocketClient::new(false, false, false, false);
        match client.connect(Some("119.147.212.81"), Some(7709)).await {
            Ok(_) => println!("Connected!"),
            Err(e) => println!("Connection error: {:?}", e),
        }
        // Rest of the program...
    }

}
